// server.cpp
#include "server.hpp"
#include "exclusive_unix_listener.hpp"
#include <sys/epoll.h>
#include <unordered_map>
#include <vector>
#include <thread>
#include <atomic>
#include <iostream>
#include <stdexcept>

namespace uds_model {

class Server::Impl {
public:
    Impl(const std::string& path)
        : path_(path), epfd_(-1), running_(false) {
        listen_fd_ = path_.native();
        set_nonblock(listen_fd_);
        epfd_ = epoll_create1(EPOLL_CLOEXEC);
        if (epfd_ == -1) throw std::system_error(errno, std::system_category(), "epoll_create1");

        epoll_event ev{};
        ev.events  = EPOLLIN | EPOLLET;
        ev.data.fd = listen_fd_;
        if (epoll_ctl(epfd_, EPOLL_CTL_ADD, listen_fd_, &ev) == -1)
            throw std::system_error(errno, std::system_category(), "epoll_ctl ADD listen");
    }

    ~Impl() {
        stop();
        close_clients();
        if (epfd_ != -1) ::close(epfd_);
    }

    void set_on_line(OnLine cb) { on_line_ = std::move(cb); }
    void set_on_connected(OnConnect cb) { on_connect_ = std::move(cb); }
    void set_on_disconnected(OnDisconnect cb) { on_disconnect_ = std::move(cb); }

    void run() {
        running_ = true;
        std::vector<epoll_event> events(64);
        while (running_.load()) {
            int nfds = epoll_wait(epfd_, events.data(), events.size(), 500);
            if (nfds == -1) {
                if (errno == EINTR) continue;
                throw std::system_error(errno, std::system_category(), "epoll_wait");
            }
            for (int i = 0; i < nfds; ++i) {
                int fd = events[i].data.fd;
                if (fd == listen_fd_) accept_all();
                else handle_client(fd, events[i].events);
            }
        }
    }

    void stop() { running_ = false; }

private:
    struct ClientCtx {
        uds::stream_socket sock;
        std::string        inbuf;
        explicit ClientCtx(uds::stream_socket s) : sock(std::move(s)) {}
    };

    exclusive_unix_listener                   path_;
    int                                       epfd_;
    int                                       listen_fd_;
    std::unordered_map<int, ClientCtx>        clients_;
    std::atomic<bool>                         running_;
    OnLine                                    on_line_;
    OnConnect                                 on_connect_;
    OnDisconnect                              on_disconnect_;

    static void set_nonblock(int fd) {
        int fl = fcntl(fd, F_GETFL, 0);
        fcntl(fd, F_SETFL, fl | O_NONBLOCK);
    }

    void close_clients() {
        for (auto& [fd, ctx] : clients_) {
            epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
            if (on_disconnect_) on_disconnect_(fd);
        }
        clients_.clear();
    }

    void accept_all() {
        while (true) {
            try {
                int cfd = path_.accept_nonblock();
                if (cfd == -1) break;

                uds::stream_socket cli{cfd};
                clients_.emplace(cfd, ClientCtx(std::move(cli)));
                
                // 如果有连接成功的回调，执行它
                if (on_connect_) on_connect_(cfd);

                epoll_event ev{};
                ev.events  = EPOLLIN | EPOLLET | EPOLLRDHUP;
                ev.data.fd = cfd;
                if (epoll_ctl(epfd_, EPOLL_CTL_ADD, cfd, &ev) == -1)
                    throw std::system_error(errno, std::system_category(), "epoll_ctl ADD client");
            } catch (const std::system_error& e) {
                if (e.code().value() == EAGAIN || e.code().value() == EWOULDBLOCK) break;
                throw;
            }
        }
    }

    void handle_client(int fd, uint32_t ev) {
        auto it = clients_.find(fd);
        if (it == clients_.end()) return;
        
        // 检测连接断开
        if (ev & (EPOLLRDHUP | EPOLLHUP)) {
            epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
            clients_.erase(it);
            if (on_disconnect_) on_disconnect_(fd);
            return;
        }
        
        // 处理数据
        if (ev & EPOLLIN) {
            char buf[4096];
            while (true) {
                ssize_t n = recv(fd, buf, sizeof(buf), MSG_DONTWAIT);
                if (n > 0) {
                    it->second.inbuf.append(buf, n);
                    std::size_t pos;
                    while ((pos = it->second.inbuf.find('\n')) != std::string::npos) {
                        std::string line = it->second.inbuf.substr(0, pos);
                        on_line_(fd, line);
                        it->second.inbuf.erase(0, pos + 1);
                    }
                } else if (n == 0) {
                    epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
                    clients_.erase(it);
                    if (on_disconnect_) on_disconnect_(fd);
                    return;
                } else {
                    if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                    epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
                    clients_.erase(it);
                    if (on_disconnect_) on_disconnect_(fd);
                    return;
                }
            }
        }
    }
};

/* ---------- Server 公有方法 ---------- */
Server::Server(const std::string& path) : pImpl(std::make_unique<Impl>(path)) {}
Server::~Server() = default;
void Server::run() { pImpl->run(); }
void Server::stop() { pImpl->stop(); }
void Server::set_on_line(OnLine cb) { pImpl->set_on_line(std::move(cb)); }
void Server::set_on_connected(OnConnect cb) { pImpl->set_on_connected(std::move(cb)); }
void Server::set_on_disconnected(OnDisconnect cb) { pImpl->set_on_disconnected(std::move(cb)); }
} // namespace uds_model
